package com.wugenqiang.test.readingFromKafka;

import com.wugenqiang.test.writeIntoOpentsdb.WriteIntoOpentsdb;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.io.IOException;
import java.util.Properties;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author createdBy liuxiaojun editBy wangzhijian and wugenqiang
 * @Title: ReadingFromKafka
 * @ProjectName kafka_flink_opentsdb
 * @Description: flink读取kafka数据
 * @date 2018/8/29 10:29
 */
public class ReadingFromKafka {

    /**
     　　* @Description: flink读取kafka数据
     　　* @param []
     　　* @return void
     　　* @throws
     　　* @author wugenqiang
     　　* @date 2018/8/29 10:33
     　　*/
    public void readingFromKafka(){

        final String ZOOKEEPER_HOST = "master:2181,slave1:2181,slave2:2181";
        final String KAFKA_HOST = "master:9092,slave1:9092,slave2:9092";
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000); // 非常关键，一定要设置启动检查点！！
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        Properties props = new Properties();
        props.setProperty("zookeeper.connect", ZOOKEEPER_HOST);
        props.setProperty("bootstrap.servers", KAFKA_HOST);
        props.setProperty("group.id", "transaction");

        DataStream<String> transaction = env.addSource(new FlinkKafkaConsumer010<String>("mastertest", new SimpleStringSchema(), props));

        transaction.rebalance().map(new MapFunction<String, Object>() {
            public String map(String value)throws IOException {

                /*//此处处理一条数据仅有一个数据项
                WriteIntoOpentsdb writeIntoOpentsdb = new WriteIntoOpentsdb();
                writeIntoOpentsdb.writeIntoOpentsdb(value);*/

                //此处处理一条数据中的多个数据项
                String result[]=value.split(" ");
                WriteIntoOpentsdb writeIntoOpentsdb = new WriteIntoOpentsdb();
                writeIntoOpentsdb.writeIntoOpentsdb(result);

                return value;
            }

        }).print();

        try {
            env.execute();
        } catch (Exception ex) {

            Logger.getLogger(ReadingFromKafka.class.getName()).log(Level.SEVERE, null, ex);
            ex.printStackTrace();
        }
    }

}
